package com.atguigu.project.app

import com.atguigu.project.bean.{CategoryCountInfo, UserVisitAction}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

/**
 * Author atguigu
 * Date 2020/10/31 14:36
 */
object CategoryTopApp {
    // 计算top10的品类
    def calcTop10Category_1(sc: SparkContext,
                            userVisitActionRDD: RDD[UserVisitAction]) = {
        // 1. 先计算每个品类的点击数
        val categoryAndClickCount = userVisitActionRDD
            .filter(_.click_category_id != -1)
            .map(action => (action.click_category_id.toString, 1))
            .reduceByKey(_ + _)
        
        // 2. 先计算每个品类的下单数数  order_category_ids这个属性是多个品类
        val categoryAndOrderCount = userVisitActionRDD
            .filter(_.order_category_ids != "null")
            .flatMap(_.order_category_ids.split(","))
            .map((_, 1))
            .reduceByKey(_ + _)
        
        // 3. 先计算每个品类的支付数据  page_category_ids这个属性是多个品类
        val categoryAndPayCount = userVisitActionRDD
            .filter(_.pay_category_ids != "null")
            .flatMap(_.pay_category_ids.split(","))
            .map((_, 1))
            .reduceByKey(_ + _)
        
        val resultRDD = categoryAndClickCount
            .fullOuterJoin(categoryAndOrderCount)
            .map {
                case (cid, (Some(clickCount), Some(orderCount))) =>
                    (cid, (clickCount, orderCount))
                case (cid, (None, Some(orderCount))) =>
                    (cid, (0, orderCount))
                case (cid, (Some(clickCount), None)) =>
                    (cid, (clickCount, 0))
            }
            .fullOuterJoin(categoryAndPayCount)
            .map {
                case (cid, (Some((clickCount, orderCount)), Some(payCount))) =>
                    (cid, (clickCount, orderCount, payCount))
                case (cid, (None, Some(payCount))) =>
                    (cid, (0, 0, payCount))
                case (cid, (Some((clickCount, orderCount)), None)) =>
                    (cid, (clickCount, orderCount, 0))
                
            }
        val result = resultRDD.sortBy {
            case (cid, (clickCount, orderCount, payCount)) => (-clickCount, -orderCount, -payCount)
        }.take(10)
        
        result.foreach(println)
    }
    
    def calcTop10Category_2(sc: SparkContext,
                            userVisitActionRDD: RDD[UserVisitAction]) = {
        
        //
        userVisitActionRDD
            .flatMap(action => {
                // 如果是点击就原封不动  [(cid1, "click"), ("cid1", "order"), ..)
                if (action.click_category_id != -1) {
                    (action.click_category_id.toString, "click") :: Nil
                } else if (action.order_category_ids != "null") {
                    action.order_category_ids.split(",").map((_, "order"))
                } else if (action.pay_category_ids != "null") {
                    action.pay_category_ids.split(",").map((_, "pay"))
                } else {
                    Nil
                }
            })
            .map((_, 1))
            .reduceByKey(_ + _)
            .map {
                case ((cid, action), count) => (cid, (action, count))
            }
            .groupByKey()
            .map {
                case (cid, it: Iterable[(String, Int)]) =>
                    // click->10 order->1  pay->1
                    val map: Map[String, Int] = it.toMap
                    (cid, (map.getOrElse("click", 0), map.getOrElse("order", 0), map.getOrElse("pay", 0)))
            }
            .sortBy {
                case (cid, (clickCount, orderCount, payCount)) => (-clickCount, -orderCount, -payCount)
            }
            .take(10)
            .foreach(println)
    }
    
    
    def calcTop10Category_5(sc: SparkContext,
                            userVisitActionRDD: RDD[UserVisitAction]) = {
        // 累加器  聚合
        // 1. 创建累加器
        val acc = new CategoryAcc
        // 2. 注册累加器
        sc.register(acc, "CategoryAcc")
        // 3. 使用累加器累加 : Map cid1-> (), cid2-> ()
        // List(,..,...,)
        userVisitActionRDD.foreach(acc.add)
        val result: List[CategoryCountInfo] = acc
            .value
            .sortBy(info => (-info.clickCount, -info.orderCount, -info.payCount))
            .take(10)
        result
    }
}

/*
.map {
                case (cid, (Some((Some(clickCount), Some(orderCount))), Some(payCount))) =>
                    (cid, (clickCount, orderCount, payCount))
                case (cid, (Some((Some(clickCount), Some(orderCount))), None)) =>
                    (cid, (clickCount, orderCount, 0))
                case (cid, (Some((Some(clickCount), None)), None)) =>
                    (cid, (clickCount, 0, 0))
                case (cid, (Some((None, Some(orderCount))), None)) =>
                    (cid, (0, orderCount, 0))
                case (cid, (None, Some(payCount))) =>
                    (cid, (0, 0, payCount))
            }
 */